package com.example.server;

import com.example.entity.SysTask;
import com.example.service.SysCuratorService;
import com.example.service.SysTaskService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Service
public class SysStandardServer implements SysServer, InitializingBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private Environment environment;

    private String serverkey;

    @Resource
    private SysCuratorService sysCuratorService;

    @Resource
    private SysTaskService sysTaskService;

    @Resource
    private DataSource dataSource;

    private List<String> serverKeysList = new ArrayList<>();

    @Override
    public void init() {
        if (!this.sysCuratorService.isExistNode(SERVER)) {
            this.sysCuratorService.createNode(CreateMode.PERSISTENT, SERVER);
        }
        if (!this.sysCuratorService.isExistNode(REFRESH)) {
            this.sysCuratorService.createNode(CreateMode.PERSISTENT, REFRESH);
        }
    }

    @Override
    public String getServerKey() {
        if (StringUtils.isEmpty(this.serverkey)) {
            if (environment.containsProperty(APP_SERVER)) {
                this.serverkey = environment.getProperty(APP_SERVER);
            } else {
                try {
                    String port = "";
                    if (environment.containsProperty(MAVEN_TOMCAT_PORT)) {
                        port = environment.getProperty(MAVEN_TOMCAT_PORT);
                    }
                    if (environment.containsProperty(SERVER_PORT)) {
                        port = environment.getProperty(SERVER_PORT);
                    }
                    this.serverkey = InetAddress.getLocalHost().getHostAddress() + COLON + port;
                } catch (UnknownHostException e) {
                    logger.error("本地ip获取失败.");
                }
            }
        }
        return this.serverkey;
    }

    @Override
    public List<String> getServerKeyList() {
        return this.serverKeysList;
    }

    /***
     *
     * 服务节点掉线回调。尝试任务转移,此处需要加入分布式悲观锁的锁定。
     */
    @Override
    public void serverOfflineHandler(String serverKey) {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        try {
            connection = this.dataSource.getConnection();
            connection.setAutoCommit(Boolean.FALSE);
            preparedStatement = connection.prepareStatement(SQL);
            preparedStatement.execute();
            List<SysTask> sysTaskList = this.sysTaskService.findTaskByOffline(serverKey);
            if (!sysTaskList.isEmpty()) {
                sysTaskService.updateAndDispatcherListToQueue(sysTaskList);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != preparedStatement) {
                try {
                    preparedStatement.close();
                } catch (SQLException e1) {
                    e1.printStackTrace();
                }
            }
            if (null != connection) {
                try {
                    connection.close();
                } catch (SQLException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }


    /****
     *
     * 通知所有的节点，清理队列，刷新数据。
     */
    @Override
    public void notifyAllServersToRefresh(Long syncTime) {
        String uuid = UUID.randomUUID().toString();
        this.sysCuratorService.createNodeAndData(CreateMode.EPHEMERAL, REFRESH + PREF + uuid, String.valueOf(syncTime));
    }


    /*
     * 功能描述: 注册本机到zk
     *
     * @Author: chentao
     * @Date: 2023/3/22 9:08
     * @return:
     */
    @Override
    public void registerServer() {
        this.sysCuratorService.createNode(CreateMode.EPHEMERAL, SERVER + PREF + getServerKey());
        this.sysCuratorService.watchNode(SERVER, new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getData() != null && !StringUtils.isEmpty(event.getData().getPath())) {
                    String path = event.getData().getPath();
                    if (path.startsWith(SERVER)) {
                        String serverKey = path.replaceFirst(SERVER + PREF, "");
                        if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                            if (!serverKeysList.contains(serverKey)) {
                                serverKeysList.add(serverKey);
                            }
                            logger.info("节点{}加入集群...", serverKey);
                        }
                        if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                            if (serverKeysList.contains(serverKey)) {
                                serverKeysList.remove(serverKey);
                            }
                            serverOfflineHandler(serverKey);
                            logger.info("节点{}已掉线通知...", serverKey);
                        }
                    }
                }
            }
        });
        this.serverKeysList = this.sysCuratorService.getNodeChild(SERVER);
        logger.warn("节点{}注册服务成功...", getServerKey());
    }

    /***
     *
     * 注册刷新事件，监听到刷新后，清空本地的队列，重新从数据库获取。
     */
    @Override
    public void registerRefreshEvent() {
        this.sysCuratorService.watchNode(REFRESH, new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                    byte[] data = event.getData().getData();
                    String rtn = new String(data);
                    sysTaskService.refreshQueue(Long.parseLong(rtn));
                }
            }
        });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.init();
        this.registerServer();
        this.registerRefreshEvent();
    }

}